package com.atguigu.day01;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

// 从文件读取数据
// word count
public class Example2 {
    // 不要忘记抛出异常！！！
    public static void main(String[] args) throws Exception {
        // 获取运行时环境，类似SparkContext
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行任务的数量为 1
        // 向资源管理器申请一个任务插槽
        env.setParallelism(1);

        // 读取数据源 -> 计算 -> 输出计算结果
        // 读取数据源
        DataStreamSource<String> stream = env
                .readTextFile("/home/zuoyuan/flink0609/src/main/resources/word.txt");

        // word count流程：map -> shuffle -> reduce
        // map: "hello world" -> ("hello", 1), ("world", 1)
        // shuffle: 将不同 key 的数据发送到不同的**逻辑分区**
        // reduce: 将同一个逻辑分区的数据聚合：[("hello", 1),("hello", 1)] -> ("hello", 2)

        // 第一步：map操作
        // 注意我们用了`flatMap`方法
        // flatMap的语义：将列表或者流中的每一个元素转换成0个、1个或者多个元素
        SingleOutputStreamOperator<Tuple2<String, Integer>> mappedStream = stream
                // 传入匿名类FlatMapFunction作为参数
                // 第一个泛型：flatMap算子输入的泛型，例如"hello world"
                // 第二个泛型：flatMap算子输出的泛型，例如("hello", 1), ("world", 1)
                .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                    // 第二个参数是集合类型，用来向下游发送数据
                    public void flatMap(String in, Collector<Tuple2<String, Integer>> out) throws Exception {
                        // 使用空格切分字符串in
                        String[] array = in.split(" ");
                        for (String word : array) {
                            // 使用`.collect`方法发出数据
                            // Tuple2.of用来实例化一个二元组
                            out.collect(Tuple2.of(word, 1));
                        }
                    }
                });

        // 第二步：shuffle操作
        // 按照key将数据分发到key对应的逻辑分区
        // 每个逻辑分区的key是一样的
        // 第一个泛型：mappedStream数据流中元素的泛型
        // 第二个泛型：key的泛型
        KeyedStream<Tuple2<String, Integer>, String> keyedStream = mappedStream
                // 第一个泛型：输出数据的类型
                // 第二个泛型：key的类型
                // 匿名类作为参数
                .keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
                    public String getKey(Tuple2<String, Integer> value) throws Exception {
                        // 指定key为元组的f0字段，f0相当于scala元组的`_1`
                        return value.f0;
                    }
                });

        // 第三步：reduce
        // 将key相同的数据，聚合起来
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = keyedStream
                // 1表示聚合的字段是元组的第1个字段，不是第0个字段
                // 每个key都有自己的sum
                // sum内部会维护一个累加器（内部状态）
                // 第一条数据来到以后，直接使用第一条数据更新累加器（因为累加器初始值为空）
                // 第二条数据以及后面的数据来到以后，和累加器进行聚合，然后更新累加器
                // sum算子向下游输出累加器
                .sum(1);

        // 输出聚合结果
        result.print();

        // 提交任务并执行任务
        env.execute();
    }
}
